package master

import (
	"distributed_db/utils"
	"distributed_db/utils/transaction"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"

	_ "github.com/go-sql-driver/mysql"
)

// FindTable 查找表，如果存在则返回表单对象对应的指针和 true，否则返回 nil 和 false
func (master *Master) FindTable(tableName string) (*Table, bool) {
	table, exists := master.tables[tableName]
	if !exists {
		return nil, false
	}
	return table, true
}

// NewTable 创建表，如果 Table 对象存在则直接返回，否则新建 Table 对象
func (master *Master) NewTable(tableName string, columns []string, primaryKeys []string, indexes []string) (*Table, error) {
	master.tablesMutex.Lock()
	defer master.tablesMutex.Unlock()
	table, exists := master.tables[tableName]
	// 如果该表已经存在，则无需创建
	if exists {
		return table, nil
	}
	// 创建新表
	table = &Table{
		TableName:  tableName,
		Columns:    columns,
		PrimaryKey: primaryKeys,
		Indexes:    indexes,
		RecordNum:  0,
		Regions:    []RegionMapping{},
	}
	master.tables[tableName] = table
	return table, nil
}

// DeleteTable 删除表
func (master *Master) DeleteTable(tableName string) error {
	master.tablesMutex.Lock()
	defer master.tablesMutex.Unlock()
	table, exists := master.tables[tableName]
	if !exists {
		return fmt.Errorf("table %s does not exist", tableName)
	}
	var wg sync.WaitGroup
	errChan := make(chan error, len(table.Regions))
	for _, region := range table.Regions {
		wg.Add(1)
		go func(r RegionMapping) {
			defer wg.Done()
			err := master.removeRegion(r.RegionName)
			errChan <- err
		}(region)
	}
	// 等待所有协程完成
	wg.Wait()
	close(errChan)
	// 处理返回值
	for res := range errChan {
		if res != nil {
			return res
		}
	}
	delete(master.tables, tableName)
	return nil
}

// BeginTransaction 开启分布式事务，返回事务 ID
func (master *Master) BeginTransaction() (string, error) {
	master.tablesMutex.Lock()
	defer master.tablesMutex.Unlock()
	trans := transaction.BeginTransaction()
	_, exists := master.transactions[trans.ID]
	if exists {
		return "", fmt.Errorf("transaction %s already exist", trans.ID)
	}
	master.transactions[trans.ID] = trans
	return trans.ID, nil
}

// CommitTransaction 提交事务
func (master *Master) CommitTransaction(transactionID string) error {
	master.tablesMutex.Lock()
	defer master.tablesMutex.Unlock()
	trans, exists := master.transactions[transactionID]
	if !exists {
		return fmt.Errorf("transaction %s does not exist", transactionID)
	}
	err := trans.Commit()
	if err != nil {
		return err
	}
	// 删除对应事务
	delete(master.transactions, transactionID)
	return nil
}

// RollbackTransaction 回滚事务
func (master *Master) RollbackTransaction(transactionID string) error {
	master.tablesMutex.Lock()
	defer master.tablesMutex.Unlock()
	trans, exists := master.transactions[transactionID]
	if !exists {
		return fmt.Errorf("transaction %s does not exist", transactionID)
	}
	err := trans.RollBack()
	if err != nil {
		return err
	}
	// 删除对应事务
	delete(master.transactions, transactionID)
	return nil
}

// GetTransaction 根据事务 ID 获取事务对象
func (master *Master) GetTransaction(transactionID string) (*transaction.Transaction, error) {
	trans, exists := master.transactions[transactionID]
	if !exists {
		return nil, fmt.Errorf("transaction %s does not exist", transactionID)
	}
	return trans, nil
}

// InsertDataToTable 插入数据到表中，并根据数据大小拆分插入到多个 Region
func (master *Master) InsertDataToTable(transactionID string, tableName string,
	columns []string, data [][]interface{}) (int, error) {
	master.tablesMutex.Lock()
	defer master.tablesMutex.Unlock()

	var wg sync.WaitGroup
	var mu sync.Mutex
	errChan := make(chan error, len(data)/RegionSize+1)
	var insertedRegion int32 = 0 // 原子操作保护并发修改

	table, exists := master.FindTable(tableName)
	if !exists {
		return 0, fmt.Errorf("table %s does not exist", tableName)
	}

	// 获取表的全局数据以检查重复
	existingData, err := master.FindDataInTable(transactionID, tableName, strings.Join(columns, ","), "")
	if err != nil {
		return 0, err
	}

	// 筛选出非重复数据
	filteredData := filterDuplicateData(existingData, columns, data)

	// 没有需要插入的数据
	if len(filteredData) == 0 {
		utils.LogC(utils.Green, "No need to insert data into table %s\n", tableName)
		return 0, nil
	}

	dataSize := len(filteredData)
	inserted := 0
	remainingData := make([][]interface{}, len(filteredData))
	copy(remainingData, filteredData)

	foundRegion := false

	for inserted < dataSize && !foundRegion {
		// 遍历已有的 Region，尝试找到一个有足够空间的 Region 进行插入
		utils.LogC(utils.Green, "inserted: %d, remaining: %d, table: %s\n", inserted, len(remainingData), tableName)
		if len(remainingData) <= RegionSize {
			var regionName string
			var regionList []*RegionMapping
			// 找到有足够空间的 Region
			for _, region := range table.Regions {
				if region.RecordNum+len(remainingData) <= RegionSize {
					regionName = region.RegionName
					foundRegion = true
					break
				}
			}
			// 找到该 Region 的所有三个副本
			for i, region := range table.Regions {
				if region.RegionName == regionName {
					regionList = append(regionList, &table.Regions[i])
				}
			}
			// 在所有副本中插入数据
			wg.Add(1)
			go func() {
				defer wg.Done()
				insertedCount, err := master.insertDataToRegion(transactionID, regionName, columns, table.PrimaryKey, table.Indexes, remainingData)
				errChan <- err
				mu.Lock()
				// 在所有的三个副本中修改数据量
				for _, r := range regionList {
					r.RecordNum += insertedCount
					table.RecordNum += insertedCount
					inserted += insertedCount
				}
				remainingData = [][]interface{}{}
				mu.Unlock()
			}()
		}

		if !foundRegion {
			// 使用原子操作获取唯一值
			regionCount := atomic.AddInt32(&insertedRegion, 1)
			// 生成唯一 Region 名称
			newRegionName := fmt.Sprintf("%s_region_%d", table.TableName, len(table.Regions)/3+int(regionCount))

			// 拆分数据
			var currentChunk [][]interface{}
			if len(remainingData) > RegionSize {
				currentChunk = splitDataToChunk(remainingData, RegionSize)
			} else {
				currentChunk = remainingData
			}
			// 更新剩余未插入的数据
			remainingData = remainingData[len(currentChunk):]

			// 开启新协程处理数据插入
			wg.Add(1)
			go func(data [][]interface{}) {
				defer wg.Done()
				// 如果没有足够的空间，创建新的 Region 并插入剩余数据
				regionMaps, err := master.addRegion(transactionID, newRegionName, columns, table.PrimaryKey, table.Indexes, data)
				errChan <- err
				if len(regionMaps) > 0 {
					mu.Lock()
					table.Regions = append(table.Regions, regionMaps...)
					table.RecordNum += len(currentChunk)
					inserted += len(currentChunk)
					mu.Unlock()
				}
			}(currentChunk)
		}

		// 如果剩余数据已全部插入，退出循环
		if len(remainingData) == 0 {
			break
		}
	}

	// 等待协程完成
	wg.Wait()
	close(errChan)
	// 检查报错信息
	e := false
	for err := range errChan {
		if err != nil {
			utils.LogC(utils.Red, "table_server insert data failed：%s\n", err.Error())
			e = true
		}
	}
	if e {
		return 0, fmt.Errorf("some data insert failed")
	}
	// 异步更新
	master.taskQueue.Enqueue(func() {
		tablesToFile(master)
	})
	return inserted, nil
}

// DeleteDataFromTable 从表中删除数据
func (master *Master) DeleteDataFromTable(transactionID string, tableName string, condition string) (int, error) {
	master.tablesMutex.Lock()
	defer master.tablesMutex.Unlock()

	table, exists := master.FindTable(tableName)
	if !exists {
		return 0, fmt.Errorf("table %s does not exist", tableName)
	}

	deleted := 0
	var wg sync.WaitGroup
	var mu sync.Mutex
	errChan := make(chan error, len(table.Regions))
	for i := range table.Regions {
		wg.Add(1)
		go func(r *RegionMapping) { // 使用参数传递
			defer wg.Done()
			count, err := master.deleteDataFromRegion(transactionID, r.RegionName, condition)
			errChan <- err
			mu.Lock()
			r.RecordNum -= count
			table.RecordNum -= count
			deleted += count
			mu.Unlock()
		}(&table.Regions[i])
	}

	wg.Wait()
	close(errChan)
	for err := range errChan {
		if err != nil {

			return deleted, err
		}
	}

	// 异步更新
	master.taskQueue.Enqueue(func() {
		tablesToFile(master)
	})

	return deleted, nil
}

// UpdateDataInTable 更新表中的数据
func (master *Master) UpdateDataInTable(transactionID string, tableName string, clause string, condition string) error {
	master.tablesMutex.Lock()
	defer master.tablesMutex.Unlock()

	table, exists := master.FindTable(tableName)
	if !exists {
		return fmt.Errorf("table %s does not exist", table.TableName)
	}

	var wg sync.WaitGroup
	errChan := make(chan error, len(table.Regions)+1)
	defer close(errChan)

	for _, region := range table.Regions {
		wg.Add(1)
		go func(r RegionMapping) {
			defer wg.Done()
			err := master.updateDataInRegion(transactionID, r.RegionName, clause, condition)
			errChan <- err
		}(region)
	}

	wg.Wait()
	for err := range errChan {
		if err != nil {
			return err
		}
	}

	return nil
}

// FindDataInTable 查找表中的数据
func (master *Master) FindDataInTable(transactionID string, tableName string,
	queryRows string, condition string) ([]map[string]interface{}, error) {
	var results []map[string]interface{}
	visited := make(map[string]bool)

	table, exists := master.FindTable(tableName)
	if !exists {
		return nil, fmt.Errorf("table %s does not exist", table.TableName)
	}

	var wg sync.WaitGroup
	var mu sync.Mutex
	errChan := make(chan error, len(table.Regions))
	// 遍历所有的 Region
	for _, region := range table.Regions {
		// 由于每个 Region 都会有三个副本，因此需要防止重复访问
		if !visited[region.RegionName] {
			wg.Add(1)
			go func(r RegionMapping) {
				defer wg.Done()
				data, err := master.findDataInRegion(transactionID, r.RegionName, queryRows, condition)
				errChan <- err
				mu.Lock()
				if err == nil && !visited[r.RegionName] {
					results = append(results, data...)
					visited[r.RegionName] = true
				}
				mu.Unlock()
			}(region)
		}
	}

	wg.Wait()
	close(errChan)
	for err := range errChan {
		if err != nil {
			return results, err
		}
	}

	return results, nil
}

// splitDataToChunk 将大数据块分割为指定大小的小数据块
func splitDataToChunk(data [][]interface{}, chunkSize int) [][]interface{} {
	if len(data) <= chunkSize {
		return data
	}
	return data[:chunkSize]
}

// filterDuplicateData 筛选出需要插入的非重复数据
func filterDuplicateData(existingData []map[string]interface{}, columns []string, data [][]interface{}) [][]interface{} {
	existingDataMap := make(map[string]bool)
	for _, row := range existingData {
		key := generateUniqueKey(row, columns)
		existingDataMap[key] = true
	}

	var filteredData [][]interface{}
	for _, row := range data {
		key := generateUniqueKeyFromRow(row, columns)
		if !existingDataMap[key] {
			filteredData = append(filteredData, row)
		}
	}
	return filteredData
}

// generateUniqueKey 根据现有数据生成唯一键
func generateUniqueKey(row map[string]interface{}, columns []string) string {
	var key strings.Builder
	for _, col := range columns {
		value := fmt.Sprintf("%s", row[col])
		key.WriteString(fmt.Sprintf("%s,", value))
	}
	return key.String()
}

// generateUniqueKeyFromRow 根据新数据生成唯一键
func generateUniqueKeyFromRow(row []interface{}, columns []string) string {
	var key strings.Builder
	for i := range columns {
		value := fmt.Sprintf("%v", row[i])
		key.WriteString(fmt.Sprintf("%s,", value))
	}
	return key.String()
}
